package com.rj.data.job.impl;

import com.google.common.collect.Lists;
import com.google.common.util.concurrent.ListenableFuture;
import com.rj.data.config.JestFactory;
import com.rj.data.job.MultiThreadJob;
import com.rj.data.task.impl.CallableFileTask;
import com.rj.data.task.lineprocessor.ComInfotoEsLineProcessor;
import com.rj.data.sink.ElasticSink;
import com.rj.data.sink.Sink;
import com.rj.data.source.FileSource;
import com.rj.data.source.Source;
import com.rj.data.task.CallableTask;
import io.searchbox.action.BulkableAction;

import java.io.File;
import java.io.FilenameFilter;
import java.io.IOException;
import java.util.List;

/**
 * 企业基本信息导入es7.5.1
 */
public class ComBaseInfoToEsJob extends MultiThreadJob<Boolean> {


    private static final String BASE_PATH = "";
    private static final String INDEX_NAME = "cominfo";

    public static void main(String[] args) {
        ComBaseInfoToEsJob comBaseInfoToEsJob = new ComBaseInfoToEsJob();
        try {
            comBaseInfoToEsJob.doTask();
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
        }
    }

    /**
     * 定义任务入口
     *
     * @return
     */
    @Override
    public List<ListenableFuture<Boolean>> defineJob() {
        List<ListenableFuture<Boolean>> futures = Lists.newArrayList();
        File dir = new File(BASE_PATH);
        File[] files = dir.listFiles(new FilenameFilter() {
            @Override
            public boolean accept(File dir, String name) {
                if (name.startsWith("base")) {
                    return true;
                }
                return false;
            }
        });
        for (File tfile : files) {
            Source source = new FileSource(tfile);
            Sink sink = new ElasticSink(INDEX_NAME, JestFactory.getJestClient());

            ComInfotoEsLineProcessor comIndextoEsLineProcessor = ComInfotoEsLineProcessor.builder()
                    .sink(sink)
                    .indexName(INDEX_NAME)
                    .build();

            List<Sink> sinkList = Lists.newArrayList(sink);
            CallableTask task = CallableFileTask.<List<BulkableAction>>callableFileTaskBuilder()
                    .source(source)
                    .sinks(sinkList)
                    .lineProcessor(comIndextoEsLineProcessor)
                    .build();
            ListenableFuture taskFu = executorService.submit(task);
            futures.add(taskFu);
        }
        return futures;
    }
}
